// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#pragma once

#include <pollux/core/i_typed_expr.h>
#include <pollux/core/query_ctx.h>
#include <pollux/plan/reference_query_runner.h>
#include <pollux/testing/expression/fuzzer_toolkit.h>
#include <pollux/functions/function_registry.h>
#include <pollux/type/type.h>
#include <pollux/vector/base_vector.h>
#include <pollux/vector/complex_vector.h>
#include <pollux/vector/vector_saver.h>
#include <pollux/vector/vector_fuzzer.h>

using namespace kumo::pollux::plan;

namespace kumo::pollux::test {

using kumo::pollux::fuzzer::InputRowMetadata;

struct ExpressionVerifierOptions {
  bool disableConstantFolding{false};
  std::string reproPersistPath;
  bool persistAndRunOnce{false};
};

class ExpressionVerifier {
 public:
  // File names used to persist data required for reproducing a failed test
  // case.
  static constexpr const std::string_view kInputVectorFileNamePrefix =
      "input_vector";
  static constexpr const std::string_view
      kInputSelectivityVectorFileNamePrefix = "input_selectivity_vector";
  static constexpr const std::string_view kInputRowMetadataFileName =
      "input_row_metadata";
  static constexpr const std::string_view kResultVectorFileName =
      "result_vector";
  static constexpr const std::string_view kExpressionSqlFileName = "sql";
  static constexpr const std::string_view kComplexConstantsFileName =
      "complex_constants";

  ExpressionVerifier(
      core::ExecCtx* execCtx,
      ExpressionVerifierOptions options,
      std::shared_ptr<ReferenceQueryRunner> referenceQueryRunner)
      : execCtx_(execCtx),
        options_(options),
        referenceQueryRunner_{referenceQueryRunner} {}

  enum class VerificationState {
    kVerifiedAgainstReference = 0,
    kBothPathsThrow = 1,
    kReferencePathUnsupported = 2,
  };
  // Executes expressions using common path (all evaluation
  // optimizations) and compares the result with either the simplified path or a
  // reference query runner. This execution is done for each input test cases
  // where the ExprSet for the common and simplified is reused between test
  // cases to simulate batches being processed via a ProjectFilter operator.
  // Returns:
  // A vector of ResultOrError objects, one for each InputTestCase. Each result
  // contains:
  //  - result of evaluating the expressions if both paths succeeded and
  //  returned the exact same vectors.
  //  - exception thrown by the common path if both paths failed with compatible
  //  exceptions. Throws otherwise (incompatible exceptions or different
  //  results).
  //  - a verification state indicating if the result was verified against the
  //  reference DB.
  std::pair<std::vector<fuzzer::ResultOrError>, std::vector<VerificationState>>
  verify(
      const std::vector<core::TypedExprPtr>& plans,
      const std::vector<fuzzer::InputTestCase>& inputTestCases,
      VectorPtr&& resultVector,
      bool canThrow,
      const InputRowMetadata& inputRowMetadata = {});

 private:
  // Utility method used to serialize the relevant data required to repro a
  // crash.
  void persistReproInfo(
      const std::vector<fuzzer::InputTestCase>& inputTestCases,
      const InputRowMetadata& inputRowMetadata,
      const VectorPtr& resultVector,
      const std::string& sql,
      const std::vector<VectorPtr>& complexConstants);

  // Utility method that calls persistReproInfo to save data and sql if
  // options_.reproPersistPath is set and is not persistAndRunOnce. Do nothing
  // otherwise.
  void persistReproInfoIfNeeded(
      const std::vector<fuzzer::InputTestCase>& inputTestCases,
      const InputRowMetadata& inputRowMetadata,
      const VectorPtr& resultVector,
      const std::string& sql,
      const std::vector<VectorPtr>& complexConstants);

 private:
  core::ExecCtx* execCtx_;
  const ExpressionVerifierOptions options_;

  std::shared_ptr<ReferenceQueryRunner> referenceQueryRunner_;
};

// Finds the minimum common subexpression which fails for a plan should it
// exist.
void computeMinimumSubExpression(
    ExpressionVerifier&& minimalVerifier,
    VectorFuzzer& fuzzer,
    const std::vector<core::TypedExprPtr>& plans,
    const std::vector<fuzzer::InputTestCase>& inputTestCases,
    const InputRowMetadata& inputRowMetadata);
} // namespace kumo::pollux::test
